package com.myrice.jms.impl;

import java.nio.ByteBuffer;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledFuture;

import org.apache.log4j.Logger;

import com.myrice.core.Connection;
import com.myrice.core.impl.DefaultSession;
import com.myrice.jms.MessageFactory;
import com.myrice.jms.MsgResponse;
import com.myrice.jms.MsgResponseCall;
import com.myrice.jms.MsgServerHandler;
import com.myrice.jms.MsgSession;

public class DefaultMsgSession extends DefaultSession implements MsgSession {
	public static final byte HEAD_RESPONSE = 0, HEAD_REQUEST = 1;

	// private static final int CAPACITY_TIMEOUT_QUEUE = 50;

	@Override
	public String toString() {
		return new StringBuilder("(").append(getInetAddress()).append("@")
				.append(getSessionId()).append(")").toString();
	}

	public int getSid() {
		int index = this.sessionId.lastIndexOf('@');
		if (index != -1) {
			return Integer.parseInt(this.sessionId.substring(index + 1));
		}
		return 0;
	}

	@Override
	public MsgServerHandler getServerHandler() {
		return (MsgServerHandler) super.getServerHandler();
	}

	public String getSessionId(Integer sid) {
		return conn.getSession().getSessionId() + "@" + sid;
	}

	public MsgSession getSession(Integer sid) {
		return getServerHandler().getSession(getSessionId(sid));
	}

	public boolean hasSessionId(Integer sid) {
		String sessionid = getSessionId(sid);
		return getServerHandler().hasSessionId(sessionid);
	}

	public void closeSession(int sid) {
		DefaultMsgSession session = (DefaultMsgSession) getSession(sid);
		if (session != null) {
			session.close();
			// log.debug(conn.getSession().getInetAddress() + "(" +
			// getSessionId()
			// + ")  closeSession(" + sid + ")");
		}
	}

	public void buildSession(int sid) {
		if (!hasSessionId(sid)) {
			createSession(getSessionId(sid));
		}
	}

	public MsgSession createSession(Integer sid) {
		if (sid == null || sid.intValue() == 0)
			throw new IllegalArgumentException(String.valueOf(sid));
		String sessionid = getSessionId(sid);
		MsgSession newSession = createSession(sessionid);
		return newSession;
	}

	public synchronized MsgSession getOrCreateSession(Integer sid) {
		MsgSession session = hasSessionId(sid) ? ((DefaultMsgSession) getSession(sid))
				.onAccpeted() : createSession(sid);
		return session;
	}

	public synchronized MsgSession createSession(String sessionId) {
		MsgSession session = getServerHandler().createSession(conn, sessionId);
		((DefaultMsgSession) session).parent = this;
		return session;
	}

	public MsgSession getParent() {
		return parent;
	}

	public MsgSession onAccpeted() {
		setClosed(false);
		updateLastTime();
		return this;
	}

	public void onClosed() {
		setClosed(true);
		clear4waiting();
	}

	public void clear4waiting() {
		if (respPool == null)
			return;
		if (respPool.size() > 0) {
			log.debug(this.getInetAddress() + " clear waiting message call: "
					+ respPool.size());
			for (MsgResponseCall response : respPool.values())
				synchronized (response) {
					response.notify();
				}
			respPool.clear();
		}
	}

	/**
	 * 
	 * @param time
	 * @return true 销毁断开连接的
	 */
	public boolean check4timeout(long time) {
		long timeout = getSessionTimeout();
		if (timeout > 0 && time - getLastTime() > timeout) {
			return onTimeout();
		}
		return false;
	}

	public long getSessionTimeout() {
		return (Long) getCoverAttributeOfUser(SESSION_KEPLIVE_TIMEOUT,
				DEFAULT_KEEPLIVE_TIMEOUT);
	}

	private long getRecvTimeout() {
		return (Long) getCoverAttributeOfUser(SESSION_RECV_TIMEOUT,
				DEFAULT_RECV_TIMEOUT);
	}

	protected boolean onTimeout() {
		if (!isClosed()) {
			updateLastTime();
			close();
			return false;
		} else {
			return true;
		}
	}

	private Object registorSync(int mode, MsgResponseCall response) {
		registor(mode, response);
		return response;
	}

	private void registor(int mode, MsgResponseCall response) {
		getRespCallWaitingPool().put(mode, response);
	}

	private MsgResponseCall unregistorSync(int mode) {
		return unregistor(mode);
	}

	private MsgResponseCall unregistor(int mode) {
		return getRespCallWaitingPool().remove(mode);
	}

	private Map<Object, MsgResponseCall> getRespCallWaitingPool() {
		if (respPool == null)
			synchronized (this) {
				if (respPool == null)
					respPool = new ConcurrentHashMap<Object, MsgResponseCall>(8);
			}
		return respPool;
	}

	public boolean send(ByteBuffer msg) {
		return send(HEAD_REQUEST, 0, msg);
	}

	public boolean send(ByteBuffer msg, MsgResponseCall response) {
		return send(msg, response, false);
	}

	public boolean send(ByteBuffer msg, MsgResponseCall response, boolean sync) {
		return send(msg, response, sync, getRecvTimeout());
	}

	public boolean send(ByteBuffer msg, MsgResponseCall response, boolean sync,
			long timeout) {
		if (response.getTimeout() != 0)
			timeout = response.getTimeout();
		int mode = nextRespId();
		if (sync) {
			try {
				Object lock = registorSync(mode, response);// 注册到响应等待池
				synchronized (lock) {
					do {
						if (send(HEAD_REQUEST, mode, msg)) {
							if (timeout == -1) {
								// ========永久等待对方响应=======

								lock.wait();
								return true;

							} else {
								// ======按超时时间等待对方响应======

								long time = System.currentTimeMillis();

								lock.wait(timeout);

								time = System.currentTimeMillis() - time;
								if (time < timeout) {
									if (respPool.containsKey(mode) == false) {
										return true;// 响应成功
									} else {
										throw new IllegalStateException(
												"connection is closed");// 被其他线程中断了响应等待。
									}
								} else {
									// 等待同步响应超时，是否继续重发消息
									if (response.onTimeout(this, msg, true,
											timeout)) {
										log.warn(getInetAddress()
												+ " Retry send call message,  sid:"
												+ mode + "  msg:" + msg);
										// 唤醒连接器，继续处理
										getServerHandler().getConnector()
												.wakeup();
									} else {
										break;// 停止继续重发消息
									}
								}
							}
						} else {
							break;
						}
					} while (true);
				}
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			unregistorSync(mode);// 注销等待响应
			log.warn(getInetAddress() + "@" + getSessionId()
					+ " wait Response timeout: mode=" + mode);
			return false;
		} else {
			registor(mode, response);
			if (timeout != -1) {
				// 启用响应超时
				if (timeout < 100)
					timeout = 100;
				TimeoutTask t = newTimeoutTask(mode, msg, (int) timeout);
				response.setTimeoutTask(t);
				ScheduledFuture<?> future = getServerHandler().schedule(t,
						timeout);
				response.setFuture(future);
			}
			return send(HEAD_REQUEST, mode, msg);
		}
	}

	private synchronized int nextRespId() {
		respCount++;
		if (respCount == 0)
			respCount = 1;
		return respCount;
	}

	/**
	 * 处理到达的响应消息
	 */
	final void onResponse(DefaultMessage message) {
		int mode = message.getMode();
		MsgResponseCall response = unregistor(mode);
		if (response == null) {
			log.warn(this.getInetAddress() + "@" + this.getSessionId()
					+ " Response is handle missed: " + message);
			return;
		}
		// 取消异步超时检测任务
		ScheduledFuture<?> s = response.getFuture();
		if (s != null) {
			response.setFuture(null);
			if (!s.cancel(false)) {
				System.err.println("Can't cancel the response timeout task.");
				return;
			} else {
				TimeoutTask t = (TimeoutTask) response.getTimeoutTask();
				if (t != null) {
					response.setTimeoutTask(null);
					t.finalize();
				}
			}
		}

		// 处理响应
		Object lock = response;
		synchronized (lock) {
			try {
				// if (getRespCallWaitingPool().remove(mode) == null) {
				// log.warn("Ignore the response of the timeout,  session: "
				// + this
				// + " mode: "
				// + mode
				// + "  resp: "
				// + response);
				// return;
				// }
				handleResponse(message, response);
			} finally {
				lock.notify();
			}
		}
	}

	private void handleResponse(DefaultMessage message, MsgResponseCall response) {
		ByteBuffer msg = message.getData();
		int status = msg.getShort();
		if (status == MsgResponse.STATUS_SUCCESS) {
			response.onSuccess(msg);
		} else {
			response.onFailed(status, msg);
		}
	}

	public TimeoutTask newTimeoutTask(int mode, ByteBuffer msg, int timeout) {
		TimeoutTask t = queue.poll();
		if (t == null)
			t = new TimeoutTask();

		t.session = this;
		t.mode = mode;
		t.msg = msg;
		t.timeout = timeout;

		return t;
	}

	private static final Queue<TimeoutTask> queue = new LinkedBlockingQueue<TimeoutTask>();

	private static class TimeoutTask implements Runnable {
		DefaultMsgSession session;
		int mode;
		ByteBuffer msg;
		int timeout;

		public void run() {
			MsgResponseCall resp = session.unregistor(mode);
			if (resp != null) {
				log.error("wait timeout by asyn response, sessionId:"
						+ session.getSessionId() + "  mode:" + mode);
				// 异步响应超时
				if (resp.onTimeout(session, msg, false, timeout)) {
					session.registor(mode, resp);
					resp.setTimeoutTask(this);
					ScheduledFuture<?> future = session.getServerHandler()
							.schedule(this, timeout);
					resp.setFuture(future);
					session.send(HEAD_REQUEST, mode, msg);
					return;// 继续重发消息
				}
			}
			finalize();
		}

		@Override
		protected void finalize() {
			// 自动回收
			mode = 0;
			timeout = 0;
			msg = null;
			session = null;
			queue.offer(this);
		}
	}

	/**
	 * 发送消息(type(byte)[,mode(int)],data(byte[...]))
	 * 
	 * @param type
	 *            0 响应消息; 1 请求消息
	 * @param mode
	 *            模块 (请求消息时，模块编号为0表示不需要响应的)
	 * @param data
	 *            消息数据
	 * @return
	 */
	public boolean send(byte type, int mode, ByteBuffer data) {
		MessageFactory f = getServerHandler().getMessageFactory();
		if (f == null) {
			throw new IllegalStateException("Not found message-factory.");
		}
		int pos = data.position();
		ByteBuffer packet = f.encode(this, type, mode, data);
		super.send(packet);
		flush();
		data.position(pos);
		return true;
	}

	public void updateLastTime() {
		lastTime = System.currentTimeMillis();
	}

	public long getLastTime() {
		return lastTime;
	}

	public DefaultMsgServerHandler getHandler() {
		return (DefaultMsgServerHandler) getServerHandler();
	}

	public DefaultMsgSession(Connection conn, String sessionId) {
		this(sessionId);
		init(conn);
	}

	public DefaultMsgSession(String sessionId) {
		super(sessionId);
	}

	/* begin members */

	private MsgSession parent;
	private int respCount;
	private long lastTime;
	private Map<Object, MsgResponseCall> respPool;

	/* end members */

	public static long DEFAULT_RECV_TIMEOUT = 10000L;
	public static long DEFAULT_KEEPLIVE_TIMEOUT = 1200000L;

	private static final Logger log = Logger.getLogger(DefaultMsgSession.class);

}
